package com.join;

import java.util.Arrays;
import java.util.HashSet;
import java.util.Set;

import com.data.ConnectedEdge;
import com.data.Node;
import com.data.TreeNode;
import com.utils.BST;
import com.utils.BST.BSTnode;
import com.utils.ConstantVariables;

/**
 * Spatial join.
 * 
 * @author bh349
 * 
 */
public class SpatialJoin {
	public static Set<ConnectedEdge> spatialJoin(Set<Node> nodes) {
		Set<ConnectedEdge> edgesBeforeUF = new HashSet<ConnectedEdge>();
		for (Node n : nodes) {
			edgesBeforeUF.add(new ConnectedEdge(n, n));
		}
		Object[] sortedNodes = nodes.toArray();
		Arrays.sort(sortedNodes);
		BST tree = new BST();
		int i = 0;
		int j = 0;
		double rangeMin = 0;
		double rangeMax = 0;

		Node a = (Node) sortedNodes[0];
		Node b = (Node) sortedNodes[0];
		Set<BSTnode> out = null;
		while (i < sortedNodes.length - 1) {
			i++;
			tree.insert(a.getNodeForRedBlackTree());
			a = (Node) sortedNodes[i];
			while (Node.isXFar(a, b)) {
				j++;
				tree.delete(b.getNodeForRedBlackTree());
				b = (Node) sortedNodes[j];
			}
			rangeMin = a.getY()
					- ConstantVariables.L_MAX[ConstantVariables.testNumber];
			rangeMax = a.getY()
					+ ConstantVariables.L_MAX[ConstantVariables.testNumber];
			// rangeMin = a.getY() - 1.5;
			// rangeMax = a.getY() + 1.5;

			out = tree.rangeFind(new TreeNode(a.getX(), rangeMin),
					new TreeNode(a.getX(), rangeMax));
			for (BSTnode node : out) {
				TreeNode tn = (TreeNode) (node.data);
				Node temp = new Node(tn.getX(), tn.getY());
				if (Node.isWithinDistance(temp, a)) {
					edgesBeforeUF.add(new ConnectedEdge(temp, a));
				}
			}
		}
		return edgesBeforeUF;
	}
}
